package com.ww.disruptor.consumer;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.ww.disruptor.event.Order;
import com.ww.disruptor.handler.Consumer;
import com.ww.disruptor.producer.Producer;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * TODO
 *
 * @author wanggw
 * @date 2024-09-15 13:40:15
 */
public class WorkHandlerDemo05 {

    public static CountDownLatch totalCountDownLatch = new CountDownLatch(100 * 100);

    public static void main(String[] args) throws InterruptedException {
        // 1、创建RingBuffer
        ExecutorService es2 = Executors.newFixedThreadPool(8);
        Disruptor<Order> disruptor = new Disruptor<>(Order::new,
                1024 * 1024,
                es2,
                ProducerType.MULTI,
                new YieldingWaitStrategy());

        RingBuffer<Order> ringBuffer = disruptor.getRingBuffer();

        Consumer[] consumers = new Consumer[10];
        for (int i = 0; i < 10; i++) {
            consumers[i] = new Consumer("C: " + i);
        }
        disruptor.handleEventsWithWorkerPool(consumers);

        disruptor.start();

        CountDownLatch latch = new CountDownLatch(1);
        for (int i = 0; i < 100; i++) {
            Producer producer = new Producer(ringBuffer);
            new Thread(() -> {
                try {
                    latch.await();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }

                for (int j = 0; j < 100; j++) {
                    producer.sendData(UUID.randomUUID().toString());
                }
            }).start();
        }

        System.out.println("线程创建完毕");
        latch.countDown();
        totalCountDownLatch.await();
        for (Consumer consumer : consumers) {
            System.out.println(consumer.getConsumerId() + "消费消息数：" + consumer.getCount());
        }
    }
}
